package com.ibm.cps.message;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.ibm.cps.message.MessageFactory.MessageType;
import com.ibm.util.exception.CPSException;

/**
 * Define Message format
 * 
 * @author Yaoliang Chen
 * 
 */
public class RawDataMessage extends DataMessage implements KryoSerializable {
	String sourceid;
	protected String schemaId;
	byte[] data;

	public RawDataMessage() {

	}

	public RawDataMessage(String tenantid, String schemaid, String sourceid,
			Comparable tsKey, byte[] data) throws CPSException {
		setUserId(tenantid);
		setSchemaId(schemaid);
		setSourceid(sourceid);
		this.data = data;
		this.tsKey = tsKey;
	}

	protected Comparable tsKey;

	public String getSchemaId() {
		return schemaId;
	}

	public void setSchemaId(String schemaId) {
		this.schemaId = schemaId;
	}

	public void setSourceid(String sourceid) {
		this.sourceid = sourceid;
	}

	public String getSourceid() {
		return sourceid;
	}

	public Comparable getTsKey() {
		return tsKey;
	}

	@Override
	public MessageType getMessageType() {
		return MessageType.JSON_DATA_MESSAGE;
	}

	public byte[] getMessageBytes() {
		return data;
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(tenantid);
		output.writeString(schemaId);
//		output.writeLong(tsKey);
		kryo.writeObject(output, tsKey);
		output.writeString(sourceid);
		output.writeInt(data.length);
		output.write(data);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		tenantid = input.readString();
		schemaId = input.readString();
//		tsKey = input.readLong();
		tsKey = kryo.readObject(input, Comparable.class);
		sourceid = input.readString();
		int length = input.readInt();
		data = input.readBytes(length);
	}
}
